package simpledb;

import java.util.*;

/**
 * The Aggregation operator that computes an aggregate (e.g., sum, avg, max,
 * min). Note that we only support aggregates over a single column, grouped by a
 * single column.
 * Aggregate给出一个迭代器，Op操作，分组的字段索引，聚合的字段索引来获得迭代器中满足Op操作的字段（元组中被分组的字段，被聚合的字段）
 */
public class Aggregate extends Operator {

    private static final long serialVersionUID = 1L;

    //迭代器
    private OpIterator child;
    //分组的字段索引
    private int aIndex;
    //聚合的字段索引
    private int gbIndex;
    //聚合的操作
    private Aggregator.Op aop;
    private TupleDesc tupleDesc;
    //Aggregator指定是IntAggregator还是stringAggregator
    private Aggregator aggregator;
    //Aggregator的迭代器
    private OpIterator iterator;
    /**
     * Constructor.
     * 
     * Implementation hint: depending on the type of afield, you will want to
     * construct an {@link IntegerAggregator} or {@link StringAggregator} to help
     * you with your implementation of readNext().
     * 
     * 
     * @param child
     *            The OpIterator that is feeding us tuples.
     * @param aIndex
     *            The column over which we are computing an aggregate.
     * @param gbIndex
     *            The column over which we are grouping the result, or -1 if
     *            there is no grouping
     * @param aop
     *            The aggregation operator to use
     */
    public Aggregate(OpIterator child, int aIndex, int gbIndex, Aggregator.Op aop) {
        this.child = child;
        this.aIndex = aIndex;
        this.gbIndex = gbIndex;
        this.aop = aop;
        this.tupleDesc = getTupleDesc();
    }

    /**
     * @return If this aggregate is accompanied by a groupby, return the groupby
     *         field index in the <b>INPUT</b> tuples. If not, return
     *         {@link simpledb.Aggregator#NO_GROUPING}
     * */
    public int groupField() {
	// some code goes here
        if(Aggregator.NO_GROUPING==gbIndex)
            return Aggregator.NO_GROUPING;
        else
            return gbIndex;
    }

    /**
     * @return If this aggregate is accompanied by a group by, return the name
     *         of the groupby field in the <b>OUTPUT</b> tuples. If not, return
     *         null;
     * */
    public String groupFieldName() {
	// some code goes here
        if(Aggregator.NO_GROUPING==gbIndex)
            return null;
        else
            return tupleDesc.getFieldName(0);
    }

    /**
     * @return the aggregate field
     * */
    public int aggregateField() {
	// some code goes here
        return aIndex;
    }

    /**
     * @return return the name of the aggregate field in the <b>OUTPUT</b>
     *         tuples
     * */
    public String aggregateFieldName() {
	// some code goes here
        return tupleDesc.getFieldName(aIndex);
    }

    /**
     * @return return the aggregate operator
     * */
    public Aggregator.Op aggregateOp() {
	// some code goes here
        return aop;
    }

    public static String nameOfAggregatorOp(Aggregator.Op aop) {
	return aop.toString();
    }

    /*
    初始化aggregator
    同时调用fillIterator（）将迭代器中元组使用aggregator..mergeTupleIntoGroup聚合
     */
    public void open() throws NoSuchElementException, DbException,
	    TransactionAbortedException {
	// some code goes here
        super.open();
        Type fieldType = null;
        if(gbIndex!=Aggregator.NO_GROUPING)
        {
            fieldType = tupleDesc.getFieldType(gbIndex);
        }
        aggregator = child.getTupleDesc().getFieldType(aIndex) == Type.INT_TYPE?new IntegerAggregator(gbIndex,fieldType,aIndex,aop,tupleDesc):
                new StringAggregator(gbIndex,fieldType,aIndex,aop,tupleDesc);
        child.open();
        fillIterator();
        iterator.open();
    }

    /**
     * Returns the next tuple. If there is a group by field, then the first
     * field is the field by which we are grouping, and the second field is the
     * result of computing the aggregate. If there is no group by field, then
     * the result tuple should contain one field representing the result of the
     * aggregate. Should return null if there are no more tuples.
     */
    protected Tuple fetchNext() throws TransactionAbortedException, DbException {
	// some code goes here
        if(iterator.hasNext())
            return iterator.next();
        else
            return null;
    }
    //填充aggregator的迭代器
    private void fillIterator() throws TransactionAbortedException, DbException {
        while(child.hasNext())
        {
            aggregator.mergeTupleIntoGroup(child.next());
        }
        iterator = aggregator.iterator();
    }

    public void rewind() throws DbException, TransactionAbortedException {
	// some code goes here
        child.rewind();
        iterator.rewind();
    }

    /**
     * Returns the TupleDesc of this Aggregate. If there is no group by field,
     * this will have one field - the aggregate column. If there is a group by
     * field, the first field will be the group by field, and the second will be
     * the aggregate value column.
     * 
     * The name of an aggregate column should be informative. For example:
     * "aggName(aop) (child_td.getFieldName(afield))" where aop and afield are
     * given in the constructor, and child_td is the TupleDesc of the child
     * iterator.

     */
    public TupleDesc getTupleDesc() {
	// some code goes here
        Type[] typeAr = null;
        String[] fieldAr = null;
        //没有指定gbIndex时，TupleDesc只有一个Field:aggreField
        if(Aggregator.NO_GROUPING==gbIndex)
        {
            typeAr = new Type[1];
            fieldAr = new String[1];
            typeAr[0] = child.getTupleDesc().getFieldType(aIndex);
            fieldAr[0] = child.getTupleDesc().getFieldName(aIndex);
        }
        //当指定gbIndex，第一个Field是gbField,第二个Field是aggreField
        else
        {
            typeAr = new Type[2];
            fieldAr = new String[2];
            typeAr[0] = child.getTupleDesc().getFieldType(gbIndex);
            fieldAr[1] = child.getTupleDesc().getFieldName(gbIndex);
            typeAr[1] = Type.INT_TYPE;
            fieldAr[1] = child.getTupleDesc().getFieldName(aIndex);
        }
	    return new TupleDesc(typeAr,fieldAr);
    }

    public void close() {
	// some code goes here
        iterator.close();
    }

    @Override
    public OpIterator[] getChildren() {
	// some code goes here
        return new OpIterator[]{child};
    }

    @Override
    public void setChildren(OpIterator[] children) {
	// some code goes here
        if(null!=children)
        {
            child = children[0];
        }
    }
    
}
